package com.huitone.smspfm.core.waterpipe.trunk;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import com.huitone.smspfm.logger.ExtLogger;

import com.huitone.smspfm.core.config.AppStaticConfigs;
import com.huitone.smspfm.core.waterpipe.branch.AbstractBranchPipe;
import com.huitone.smspfm.redis.model.SmsEntity;

public abstract class AbstractMainPipe implements IMainPipe {
	
	private static ExtLogger logger = ExtLogger.create(AbstractMainPipe.class);

	private List<AbstractBranchPipe> branchPipes = null;
	
	private Date mpStartTime = null;
	//private Date mpEndTime = null;
	
	private static long lastTime = System.currentTimeMillis();
	
	private static long lastCount = 0;
	
	public void addBranchPipe(AbstractBranchPipe branchPipe) {
		
		if (branchPipe == null) return;
		
		if (branchPipes == null) {
			branchPipes = new ArrayList<>();
		}
		
		branchPipes.add(branchPipe);
	}
	
	public void addBranchPipes(List<AbstractBranchPipe> branchPipes) {
		this.branchPipes = branchPipes;
	}
	
	public void process(SmsEntity entity) {
		flow(entity);
	}
	
	public void process(List<SmsEntity> entities) {
		
		mpStartTime = new Date();
		
		flow(entities);
		
	}
	
	private void flow(List<SmsEntity> entities) {
		
		List<SmsEntity> tempList = null;
		
		if (branchPipes != null && entities != null) {
			
			for (SmsEntity entity:entities) {
				
				if (entity == null) continue;
				
				if (AppStaticConfigs.speedStatisticsOn) {
					speeds();
				}
				
				for (AbstractBranchPipe branchPipe:branchPipes) {
					branchPipe.setMpStartTime(mpStartTime);
					if ((entity = branchPipe.filter(entity)) == null) break;
				}
				
				if (entity != null) {
					if (tempList == null) tempList = new ArrayList<>();
					tempList.add(entity);
				} 
			}
		} else {
			tempList = entities;
		}
		
		lastProcess(tempList);
	}
	
	private void flow(SmsEntity entity) {
		
		if (branchPipes != null && entity != null) {
			
			for (AbstractBranchPipe branchPipe:branchPipes) {
				branchPipe.setMpStartTime(mpStartTime);
				if ((entity = branchPipe.filter(entity)) == null) break;
			}
			
			if (entity != null) {
				lastProcess(entity);
			} 
		}
	}
	
	public synchronized void speeds() {
		
		lastCount += 1;
		
		long currentTimeMillis = System.currentTimeMillis();
		
		if ((currentTimeMillis - lastTime) >= 1000) {
			logger.fatal("下行短信处理速度(条/s):" + lastCount + " poolSize:" + AppStaticConfigs.mtsmsExcutePoolSize);
			lastCount = 0;
			lastTime = System.currentTimeMillis();
		}
		
	}
	
	protected abstract void lastProcess(List<SmsEntity> entities);
	
	protected abstract void lastProcess(SmsEntity entity);
}
